package ontime

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @author dinghao 
  * @create 2021-07-28 9:58 
  * @message
  */
object SessionWindowsflink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val socketData = env.socketTextStream(args(0),9999)
    socketData.flatMap(_.split(" ")).map((_,1)).keyBy(_._1)
      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
      .reduce((a,b)=>(a._1,a._2+b._2)).print()
    env.execute("SessionWindowsflink")
  }
}
